﻿using Confluent.Kafka.Admin;
using Confluent.Kafka;
using LJ.EventBus.FastDog.Data;
using LJ.EventBus.FastDog.Transport.Abstractions;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Threading;

namespace LJ.EventBus.FastDog.Transport.Kafka.Internal;

public class KafkaConsumerClient(string groupId, EventBusTransportConfig eventBusTransportConfig) : IConsumerClient
{
    private static readonly object Lock = new();
    private readonly string _groupId = groupId;
    private readonly EventBusTransportConfig _eventBusTransportConfig = eventBusTransportConfig;
    private IConsumer<string, byte[]>? _consumerClient;

    public Func<TransportMessage, object, Task>? OnMessageCallback { get; set; }
    public Action<LogMessageEventArgs>? OnLogCallback { get; set; }

    public ICollection<string> FetchTopics(IEnumerable<string> topicNames)
    {
        ArgumentNullException.ThrowIfNull(topicNames);

        try
        {
            var config = new AdminClientConfig(_eventBusTransportConfig.MainConfig) { BootstrapServers = _eventBusTransportConfig.Servers };

            using var adminClient = new AdminClientBuilder(config).Build();

            adminClient.CreateTopicsAsync(topicNames.Select(x => new TopicSpecification
            {
                Name = x,
                NumPartitions = _eventBusTransportConfig.Topic.NumPartitions,
                ReplicationFactor = _eventBusTransportConfig.Topic.ReplicationFactor
            })).GetAwaiter().GetResult();
        }
        catch (CreateTopicsException ex) when (ex.Message.Contains("already exists"))
        {
        }
        catch (Exception ex)
        {
            var logArgs = new LogMessageEventArgs
            {
                LogType = MqLogType.ConsumeError,
                Reason = "An error was encountered when automatically creating topic! -->" + ex.Message
            };
            OnLogCallback!(logArgs);
        }

        return topicNames.ToList();
    }

    public void Listening(TimeSpan timeout, CancellationToken cancellationToken)
    {
        Connect();

        while (!cancellationToken.IsCancellationRequested)
        {
            ConsumeResult<string, byte[]> consumerResult;

            try
            {
                consumerResult = _consumerClient!.Consume(timeout);

                if (consumerResult == null) continue;
                if (consumerResult.IsPartitionEOF || consumerResult.Message.Value == null) continue;
            }
            catch (ConsumeException e) when (_eventBusTransportConfig.RetriableErrorCodes.Contains(e.Error.Code))
            {
                var logArgs = new LogMessageEventArgs
                {
                    LogType = MqLogType.ConsumeRetries,
                    Reason = e.Error.ToString()
                };
                OnLogCallback!(logArgs);

                continue;
            }

            Consume(consumerResult).GetAwaiter().GetResult();
        }
    }

    public void Subscribe(IEnumerable<string> topics)
    {
        ArgumentNullException.ThrowIfNull(topics);

        Connect();

        _consumerClient!.Subscribe(topics);
    }


    public void Reject(object? sender)
    {
        _consumerClient!.Assign(_consumerClient.Assignment);
    }

    public void Commit(object? sender)
    {
        _consumerClient!.Commit((ConsumeResult<string, byte[]>)sender!);
    }

    public void Dispose()
    {
        _consumerClient?.Dispose();
        GC.SuppressFinalize(this);
    }


    private void Connect()
    {
        if (_consumerClient != null) return;

        lock (Lock)
        {
            if (_consumerClient == null)
            {
                var config = new ConsumerConfig(new Dictionary<string, string>(_eventBusTransportConfig.MainConfig));
                config.BootstrapServers ??= _eventBusTransportConfig.Servers;
                config.GroupId ??= _groupId;
                config.AutoOffsetReset ??= AutoOffsetReset.Earliest;
                config.AllowAutoCreateTopics ??= true;
                config.EnableAutoCommit ??= false;
                config.LogConnectionClose ??= false;

                _consumerClient= new ConsumerBuilder<string, byte[]>(config)
                .SetErrorHandler(ConsumerClient_OnConsumeError)
                .Build();
            }
        }
    }

    private void ConsumerClient_OnConsumeError(IConsumer<string, byte[]> consumer, Error e)
    {
        var logArgs = new LogMessageEventArgs
        {
            LogType = MqLogType.ServerConnError,
            Reason = $"An error occurred during connect kafka --> {e.Reason}"
        };
        OnLogCallback!(logArgs);
    }


    private async Task Consume(ConsumeResult<string, byte[]> consumerResult)
    {
        var headers = new Dictionary<string, string?>(consumerResult.Message.Headers.Count);
        foreach (var header in consumerResult.Message.Headers)
        {
            var val = header.GetValueBytes();
            headers[header.Key] = val != null ? Encoding.UTF8.GetString(val) : null;
        }

        headers[MessageHeaders.Group] = _groupId;

        var message = new TransportMessage(headers, consumerResult.Message.Value);

        await OnMessageCallback!(message, consumerResult);
    }

}
